package start.redisstream;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import start.util.SpringContextUtil;

import java.util.Map;

@Slf4j
public class ProductUpdateStreamListener implements StreamListener<String, MapRecord<String, String, String>> {

    private String customerName;

    private String group;

    int count = 0;

    public ProductUpdateStreamListener(String customerName, String group) {
        this.customerName = customerName;
        this.group = group;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        StringRedisTemplate redisTemplate = (StringRedisTemplate) SpringContextUtil.getBean("stringRedisTemplate");
        RecordId recordId = message.getId();
        String stream = message.getStream();
        Map<String, String> record = message.getValue();
        /*try {
            Thread.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }*/
        //先删除，在消费
        count++;
        redisTemplate.opsForStream().delete(stream, recordId.getValue());
        log.info("[自动] customer[{}] group:[{}] 接收到一个消息 stream:[{}],id:[{}],value:[{}],当前消费总记录数[{}]", customerName, group, stream, recordId, record, count);
    }
}
